package cn.edu.flink.scala.tutorial.transformer

import cn.edu.flink.scala.tutorial.source.Event
import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala._

object 富函数方法 {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()
    env.setParallelism(2)

    val events = List(
      Event("Mary", "home/", 10000L),
      Event("Mary", "cart/", 20000L))

    val eventsDS = env.fromCollection(events)

    eventsDS.map(new MyRichMapFunction()).print("map")

    env.execute("基本转换算子")
  }

  class MyRichMapFunction extends RichMapFunction[Event, String] {
    override def map(in: Event): String = in.toString

    override def open(parameters: Configuration): Unit = println(s"任务${getRuntimeContext.getIndexOfThisSubtask}开始")

    override def close(): Unit = println(s"任务${getRuntimeContext.getIndexOfThisSubtask}结束")
  }
}
